package com.yh.csx.bsf.ods.fullload;

import com.yh.csx.bsf.core.base.Ref;
import com.yh.csx.bsf.core.db.DbConn;
import com.yh.csx.bsf.core.thread.ThreadPool;
import com.yh.csx.bsf.core.util.*;
import com.yh.csx.bsf.ods.AbstractCanalServerProvider;
import com.yh.csx.bsf.ods.SafeProperties;
import com.yh.csx.bsf.ods.meta.OdsSyncAdapter;

import lombok.var;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.File;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Component
public class CanalServerProvider extends AbstractCanalServerProvider {

	@Autowired
	private TaskManager taskManager;
	@Autowired
	Map<String, OdsSyncAdapter> syncers;
	private ExecutorService executorService = Executors
			.newFixedThreadPool(PropertyUtils.getPropertyCache("fullload.parallelCount", 10) * 2);

	private Ref<Short> clientId=new Ref((short)0);
	@Override
	public void start() {
		stop();
		// 解压文件夹
		unzip(true);
		LogUtils.info(this.getClass(), "ods", "canal server 已解压!");
		ThreadPool.System.submit("ods-canal监听任务执行", () -> {
			while (!ThreadPool.System.isShutdown()) {
				try {
					Thread.sleep(10 * 1000);
				} catch (Exception e) {
				}
				try {
					monitorTask();
				} catch (Exception e) {
					LogUtils.error(this.getClass(), "ods", "【警告】ods-canal监听任务执行出错", e);
				}
			}
		});
		// 暂缓一下
		try {
			Thread.sleep(5 * 1000);
		} catch (Exception e) {
		}
		super.start();
	}

	private void monitorTask() {
		long runningCount = taskManager.getRunningCount();
		long waitingCount = taskManager.getWaitingCount();
		if (waitingCount > 0) {
			long parallelCount = PropertyUtils.getPropertyCache("fullload.parallelCount", 1);
			long newCount = parallelCount - runningCount;
			if (newCount > 0) {
				// 设置进度
				List<TaskManager.Task> newTasks = taskManager.setRunning(newCount);
				StringBuilder sbDestinations = new StringBuilder();
				newTasks.forEach(c -> {
					c.getSubTasks().values().forEach(s -> {
						s.setProcess(0L);
						var destination = s.getDb() + "." + s.getTable();
						sbDestinations.append(destination + ",");
					});
					refreshProperty(c.getSubTasks().values());
				});
				String destinations = StringUtils.strip(sbDestinations.toString(), ",");
				LogUtils.info(this.getClass(), "ods", "【准备开启任务】:" + destinations);
				// 更新canalserver destinations 配置
				newTasks.forEach(c -> {
					final CountDownLatch subFullTaskCount = new CountDownLatch(c.getSubTasks().size());
					List<CanalFullLoadConsumer> newConsumers = new ArrayList<CanalFullLoadConsumer>();
					executorService.submit(() -> {						
						c.getSubTasks().values().forEach(s -> {// 分表多线程处理
							executorService.submit(() -> {
								clientId.setData((short)(clientId.getData().shortValue()+1));
								CanalFullLoadConsumer consumer = new CanalFullLoadConsumer();
								consumer.setAddress("127.0.0.1:11111");
								consumer.setDebug(true);
								consumer.setDestination(s.getDb() + "_" + s.getTable());
								consumer.setFilter(s.getDb() + "\\." + s.getTable());
								consumer.setUsername("");
								consumer.setPassword("");
								consumer.setClientId(clientId.getData());
								consumer.setTask(s);
								consumer.setStragtegyProvider(stragtegyProvider);
								consumer.setSyncers(syncers);
								consumer.setBatchSize(stragtegyProvider.getCanalBatchSize());
								if (TaskManager.consumers.putIfAbsent(s.getDb() + "_" + s.getTable(),
										consumer) == null) {
									newConsumers.add(consumer);
									try {// 物理表同步
										consumer.fullLoad();
									} catch (Throwable e) {
										LogUtils.error(this.getClass(), "ods", "【批量同步异常】:" + destinations, e);
									}
								}
								subFullTaskCount.countDown();
							});
						});

						try {// 集合所有子表同步结束
							subFullTaskCount.await();
						} catch (InterruptedException e) {
							LogUtils.error(this.getClass(), "ods", "【批量同步异常】:" + destinations, e);
						}
						// 重新命名表
						CanalFullLoadConsumer first = newConsumers.get(0);
						if (first != null) {
							first.renameTempToOrginal();
						}

						// 订阅增量
						try {// 暂停10s后在启动canal处理
							Thread.sleep(10000);
						} catch (InterruptedException e) {
						}
						newConsumers.forEach(d -> {// 启动增量追加
							ThreadPool.System.submit("fullload-canal", () -> {
								d.start();
							});
						});
					});
				});
				LogUtils.info(this.getClass(), "ods", "【已启动canal 批量消费者】:" + destinations);
				// 暂缓一下
				try {
					Thread.sleep(5 * 1000);
				} catch (Exception e) {
				}
			}
		}
	}

	private void refreshProperty(Collection<TaskManager.SubTask> tasks) {
		// 配置canal.properties
		/*
		 * SafeProperties sp = new SafeProperties(); var file = new
		 * File("canal/conf/canal.properties"); sp.load(file);
		 * sp.setProperty("canal.id", NetworkUtils.getIpAddress().replace(".", ""));
		 * sp.setProperty("canal.instance.filter.query.dcl", "true");
		 * sp.setProperty("canal.instance.filter.query.dml", "true");
		 * sp.setProperty("canal.instance.filter.query.ddl", "true");
		 * sp.setProperty("canal.instance.tsdb.enable", "false");
		 */

		Set<String> destinations = new HashSet<String>();
		// 配置实例
		for (var task : tasks) {
			var destination = task.getDb() + "_" + task.getTable();
			destinations.add(destination);
			FileUtils.deleteDir(new File("canal/conf/" + destination));
			// 复制文件夹
			FileUtils.copyDir("canal/example", "canal/conf/" + destination);
			try (DbConn dbConn = stragtegyProvider.getDb(task.getDb())) {
				if (dbConn == null)
					throw new RuntimeException("未配置数据库数据源:" + task.getDb());
				var dbinfo = stragtegyProvider.getDbInfo(task.getDb());
				var rs = dbConn.executeList("show master status", null);
				if (rs.size() == 0) {
					throw new RuntimeException("数据库不支持主从同步");
				}
				var journalName = rs.get(0).getOrDefault("File", "");
				var position = rs.get(0).getOrDefault("Position", "");
				SafeProperties sp2 = new SafeProperties();
				var file2 = new File("canal/conf/" + destination + "/instance.properties");
				sp2.load(file2);
				sp2.setProperty("canal.instance.filter.regex", task.getDb() + "\\." + task.getTable());
				sp2.setProperty("canal.instance.tsdb.enable", "false");
				sp2.setProperty("canal.instance.master.address", dbinfo.get("address").replace("\\", ""));
				sp2.setProperty("canal.instance.master.journal.name", ConvertUtils.convert(journalName, String.class));
				sp2.setProperty("canal.instance.master.position", ConvertUtils.convert(position, String.class));
				sp2.setProperty("canal.instance.dbUsername", dbinfo.get("username"));
				sp2.setProperty("canal.instance.dbPassword", dbinfo.get("password"));
				sp2.store(file2);
			}
		}
		// sp.setProperty("canal.destinations", StringUtils.join(destinations, ","));
		// sp.store(file);
	}

}
